DevIOの新着記事がメディアポリシーに準拠しているかBedrockで自動レビューしてみた

#Amazon Bedrock
岩田智哉
2024.09.11
リテールアプリ共創部@大阪の岩田です。
DevelopersIOにはメディアポリシーが存在します。簡単に紹介するとNDAを遵守するとか著作権法を遵守するとか、そういった内容がポリシーとして定められています。しかし、これらのポリシーは全てが定量的に評価できる指標ではなく、どうしても人によって受け取り方が異なるような定性的な指標も存在します。
例えばですが以下のように自分は褒めているつもりでも相手は不快に感じるような表現も考えられます。
（例：記事「〇〇さんは最高にイケてるおじさんです！」 → 〇〇「俺は…おじさん…」）
こういった表現についてはどうしてもセルフチェックだけでは気付き辛い部分があるので、対策のためDevIOの新着記事がメディアポリシーに準拠しているかBedrockに自動レビューしてもらう環境を作ってみました。
 構成今回作成するシステムの概要です。ざっくりこんな構成を作ります。
EventBridge Schedularを利用してLambdaを起動
DevIOの新着記事のURLを取得してSQSに送信する

SQSからLambdaを起動
新着記事のURLから記事の本文を取得し、Bedrockにレビューしてもらう
レビュー結果をSlackに通知する

 環境今回利用した環境は以下の通りです
Python: 3.12
beautifulsoup4: 4.12.3
boto3: 1.35.15
feedparser:6.0.11
markdownify: 0.13.1
requests:2.32.3
slack-sdk: 3.32.0
 実装ここからは実装を紹介していきます
 新着記事の一覧を取得するLambdaこちらの記事で紹介されていたコードを流用させてもらいました
https://dev.classmethod.jp/articles/openai-developersio-slack/
blog-feed/app.py
import boto3
from datetime import datetime, timedelta, timezone
import feedparser
import os

sqs = boto3.client('sqs')
JST = timezone(timedelta(hours=+9))
feed_url = 'https://dev.classmethod.jp/feed/'
queue_url = os.environ['QUEUE_URL']

def get_feed_entries():
    updated_since = datetime.now(JST) - timedelta(hours=1)
    feed = feedparser.parse(feed_url)
    new_entries = [
        entry for entry in feed.entries
        if datetime(*entry.updated_parsed[:6], tzinfo=timezone.utc)
        .astimezone(JST) > updated_since
    ]
    return new_entries

def lambda_handler(event, context):

    new_entries = get_feed_entries()
    entries = [
        {
            "Id": entry["id"],
            "MessageBody": entry['link']
        } for entry in  new_entries
    ]
    if len(entries) == 0:
        print('新着記事が見つからなかったため処理を終了します')
        return
    sqs.send_message_batch(
        QueueUrl=queue_url,
        Entries=entries
    )
 記事の内容をレビューしてもらうLambdaまずSQSから起動するLambdaのhandler部分です。
SSMパラメータストアからSlackのトークンを取得する等、諸々の初期処理を実行してからreview_and_notify_slackというメイン処理を呼び出しています。SQSのメッセージが重複しても実害がないため簡略化のため重複チェックは実施していません。
blog-review/app.py
from aws_lambda_powertools import Logger
import boto3
import os
from slack_sdk import WebClient

from functions import review_and_notify_slack

logger = Logger()
ssm = boto3.client("ssm")

ssm_res = ssm.get_parameter(Name="/blog-review/prd/slack-bot-token", WithDecryption=True)
slack_token = ssm_res['Parameter']['Value']
slack_channel = os.environ["SLACK_CHANNEL_ID"]
slack_client = WebClient(token=slack_token)
bedrock_runtime = boto3.client("bedrock-runtime")
system_prompt = '''
あなたは企業ブログのレビュワーです

ブログ内に不適切な表現がないかチェックする必要があります。

...略
'''

def lambda_handler(event, context):

    records = event['Records']
    batch_item_failures = []
    response = {}

    for record in records:

        try:
            review_and_notify_slack(
                url=record['body'],
                system_prompt=system_prompt,
                bedrock_runtime=bedrock_runtime,
                slack_client=slack_client,
                slack_channel=slack_channel,
            )
        except Exception as e:
            logger.error({
                "message": "Failed to process record",
                "record": record,
                "error": str(e)
            })
            batch_item_failures.append({"itemIdentifier": record['messageId']})

    response["batchItemFailures"] = batch_item_failures
    return response
レビュー&Slack投稿するためのメイン処理は以下の通りです。review_and_notify_slackという関数の中で諸々実行しています。テストしやすくなるかなと思いレビュー対象記事のURLやプロンプトを引数で渡せるようにしたのですが、結局このあたりってモックを使ったユニットテストでは意義が薄いし、かといってレビュー結果は機会的にassertできないし、どういう風に処理を分割するのが良いんでしょうね？まあ個人利用するツールみたいな位置づけなので、今回はあまり深く考えないようにします。
blog-review/functions.py
import json
import requests
from bs4 import BeautifulSoup
from markdownify import markdownify
from mypy_boto3_bedrock_runtime.client import BedrockRuntimeClient
from slack_sdk import WebClient

def review_and_notify_slack(
    url: str,
    system_prompt: str,
    bedrock_runtime: BedrockRuntimeClient,
    slack_client: WebClient,
    slack_channel: str,
):
    res = requests.get(url)
    soup = BeautifulSoup(res.text, "html.parser")
    article = soup.find("article")
    md_article = markdownify(article.prettify())

    model_id = "anthropic.claude-3-haiku-20240307-v1:0"
    review_content = f"""
以下ブログのレビューお願いします

```
{md_article}
```
    """

    res = bedrock_runtime.invoke_model(
        modelId=model_id,
        body=json.dumps(
            {
                "anthropic_version": "bedrock-2023-05-31",
                "max_tokens": 2000,
                "system": system_prompt,
                "messages": [
                    {
                        "role": "user",
                        "content": [{"type": "text", "text": review_content}],
                    }
                ],
            }
        ),
    )

    response_body = json.loads(res.get("body").read())
    print(response_body)

    slack_client.chat_postMessage(
        channel=slack_channel,
        blocks=[
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": "以下のブログをレビューしました"},
            },
            {"type": "section", "text": {"type": "mrkdwn", "text": url}},
            {"type": "divider"},
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": response_body["content"][0]["text"]},
            },
            {"type": "divider"},
        ],
    )
 諸々のリソースをデプロイするSAMテンプレート今回は上記のLambdaや関連するリソースをSAMテンプレートで定義しました。テンプレートは以下の通りです。
template.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Parameters:
  SlackChannelId:
    Type: String
    Description: Slack Channel ID
Globals:
  Function:
    Runtime: python3.12
    Timeout: 300
    MemorySize: 128
    Tracing: Active
  Api:
    TracingEnabled: true
Resources:
  BlogUrlQueue:
    Type: AWS::SQS::Queue
    Properties:
      VisibilityTimeout: 330
  BlogUrlQueuePolicy: 
    Type: AWS::SQS::QueuePolicy
    Properties: 
      PolicyDocument: 
        Statement: 
          - Sid: allo-sqs
            Effect: Allow
            Principal: "*"
            Action: "SQS:*"
            Resource: !GetAtt BlogUrlQueue.Arn
      Queues:
        - !Ref BlogUrlQueue
  BlogFeedFunc:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: blog-feed/
      Handler: app.lambda_handler
      Architectures:
      - x86_64
      Timeout: 10
      Environment:
        Variables:
          QUEUE_URL: !GetAtt BlogUrlQueue.QueueUrl
      Events:
        BlogFeedSchedule:
          Type: Schedule
          Properties:
            Schedule: cron(0 * * * ? *)
            Enabled: true
  BlogReviewFunc:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: blog-review/
      Handler: app.lambda_handler
      Architectures:
      - x86_64
      Role: !GetAtt BlogReviewFuncRole.Arn
      Events:
        SQSEvent:
          Type: SQS
          Properties:
            Queue: !GetAtt BlogUrlQueue.Arn
            BatchSize: 10
            FunctionResponseTypes:
              - ReportBatchItemFailures            
      Environment:
        Variables:
          SLACK_CHANNEL_ID: !Ref SlackChannelId
  BlogReviewFuncRole:
    Type: "AWS::IAM::Role"
    Properties:
      RoleName: bedrock-slack-backlog-rag-app-lambda-role
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      Policies:
        - PolicyName: allow-bedrock
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action:
                  - bedrock:InvokeAgent
                  - bedrock:InvokeModel
                  - bedrock:Retrieve
                  - bedrock:InvokeModelWithResponseStream
                Resource: "*"
        - PolicyName: allow-ssm
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action:
                  - ssm:GetParameter
                Resource: "*"                
        - PolicyName: allow-logs
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
            - Effect: Allow
              Action:
              - logs:CreateLogGroup
              - logs:CreateLogStream
              - logs:PutLogEvents
              Resource: '*'
レビュー結果を通知する対象のSlackチャンネルIDはパラメータで受け取るようにしているのでデプロイ時に sam deploy --parameter-overrides SlackChannelId=xxxxxxのようにparametersオプションを指定してください。
上記SAMテンプレートのデプロイに加え、Slackの投稿用に使うボットのトークンをSSMパラメータストアに登録しておいてください。
aws ssm  put-parameter --name '/blog-review/prd/slack-bot-token' --value 'Slackのボット用トークン' --type SecureString
Slack側の設定などはすでに多数のブログが存在するため割愛します。Botにchat:writeの権限だけ付与しておいてください。
 実行結果デプロイ後にSlackのメッセージを確認してみましょう。
うまく動作していそうです。
ちなみに記事の内容をレビューするLambdaはSQSから起動するので、SQSにブログのURLさえ送信すれば新着記事取得の定期実行を待たずしてレビュー自動レビューさせることも可能です。せっかくなのでDevIOのブログの中でも人気の高い？以下ブログのURLを手動でSQSに送信してBedrockにレビューしてもらいましょう。
https://dev.classmethod.jp/articles/reinvent2017-report-what-do-you-do-if-you-lost-your-id/
「パスポート紛失時の対応方法を詳細に解説した非常に参考になる記事である」とのことです。さすがです！！
 まとめDevIOの記事レビューを自動化してみました。考え方自体は類似の用途に色々と展開できるのではないでしょうか？
実行結果を見る限りツールとしてうまく動作してくれてそうではありますが、実際にメディアポリシーに違反した記事が公開されない限りレビューの有効性を評価できないのが難しいポイントだなと感じました。Bedrockのプレイグラウンドでメディアポリシーをプロンプトに設定してテストする分にはきちんと不適切な文章に対して指摘してくれていたので、恐らく有効に機能してくれているのでは...と期待しています。
しばらく運用を続けてみて様子をみたいと思います。
 参考ソースコードの一部など以下のブログを参考にさせて頂きました
OpenAIを使ってDevelopersIOの新着記事を3行で要約してSlackに投稿してみた | DevelopersIO
【Python】Beautiful Soup を使ってブログ記事のテキストを抜き出してみる | DevelopersIO